package com.xrui.hbase.codec.impl;

import com.google.common.base.Preconditions;

import com.xrui.hbase.DecodedCell;
import com.xrui.hbase.codec.CellDecoder;
import com.xrui.hbase.impl.BoundColumnReaderSpec;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.impl.AvroCellSchema;
import com.xrui.hbase.schema.spec.CellSpec;
import com.xrui.hbase.util.ByteStreamArray;
import org.apache.avro.Schema;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;

import java.io.IOException;
import java.nio.ByteBuffer;

/**
 * Base class for decoders that read HBase cells encoded using Avro.
 * <p>
 * HBase cells are encoded as Extension records, although for efficiency reasons, this class
 * provides a custom decoder that avoid copying bytes unnecessarily.
 *
 * @param <T> The type of the decoded cell data.
 */
public abstract class AvroCellDecoder<T> implements CellDecoder<T> {

    /**
     * <p>
     * Reader Avro schema. Null means use the writer schema.
     * </p>
     * <p>
     * May currently be null if the specific class for an Avro record is not available.
     * This behavior is not intended and will be removed in the future.
     * </p>
     * </p>
     * <p>
     * With SCHEMA-295, users will be able to override the reader schema.
     * Null allows a user to request the actual writer schema.
     * </p>
     */
    private final Schema mReaderSchema;

    /**
     * Initializes an abstract AvroCellDecoder.
     *
     * @param cellSpec Specification of the cell encoding.
     * @throws IOException on I/O error.
     */
    protected AvroCellDecoder(CellSpec cellSpec) throws IOException {
        Preconditions.checkNotNull(cellSpec);
        Preconditions.checkArgument(cellSpec.isAvro());
        mReaderSchema = cellSpec.getAvroSchema();
    }

    // -----------------------------------------------------------------------------------------------

    /**
     * Initializes an abstract AvroCellDecoder.
     *
     * @param layout TableLayout from which to create storage information.
     * @param spec   Specification of the cell encoding.
     * @throws IOException on I/O error.
     */
    protected AvroCellDecoder(TableLayout layout, BoundColumnReaderSpec spec) throws IOException {
        mReaderSchema = getReaderSchema(layout, spec);
    }

    /**
     * Get the Avro Schema to use to decode cells.
     *
     * @param layout TableLayout from which to create storage information.
     * @param spec   Specification of the cell encoding.
     * @return the Avro Schema to use to decode cells.
     * @throws IOException in case the specified column does not exist.
     */
    private static Schema getReaderSchema(TableLayout layout, BoundColumnReaderSpec spec)
        throws IOException {

        final AvroCellSchema cellSchema = (AvroCellSchema) layout.getCellSchema(spec.getColumnName());
        Preconditions.checkNotNull(cellSchema);
        return layout.getSchemaResolver().apply(cellSchema.getSchemaID());

    }

    /**
     * Factory for DatumReader instances.
     * <p>
     * Sub-classes must create DatumReader implementations for specific or generic records.
     *
     * @param writer Writer schema.
     * @param reader Reader schema.
     * @return a new DatumReader instance for the specified writer/reader schema combination.
     */
    protected abstract DatumReader<T> createDatumReader(Schema writer, Schema reader);

    /**
     * {@inheritDoc}
     */
    @Override
    public DecodedCell<T> decodeCell(byte[] encodedBytes) throws IOException {
        return decode(encodedBytes, null);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public T decodeValue(byte[] bytes) throws IOException {
        return decodeCell(bytes).getData();
    }

    /**
     * Decodes the serialized bytes into a DecodedCell.
     *
     * @param bytes The bytes from an HBase table cell.
     * @param reuse If non-null, may be filled with the decoded data and used as the data payload in
     *              the return value.
     * @return The decoded HBaseCell.
     * @throws IOException If there is an error.
     */
    private DecodedCell<T> decode(byte[] bytes, T reuse) throws IOException {
        final ByteStreamArray byteStream = new ByteStreamArray(bytes);
        final ByteBuffer binaryData =
            ByteBuffer.wrap(bytes, byteStream.getOffset(), bytes.length - byteStream.getOffset());
        final T data = decodeAvro(binaryData, mReaderSchema, mReaderSchema, reuse);
        return new DecodedCell<>(data);
    }

    /**
     * Decodes the data payload given the reader and writer schema. If reuse is non-null, the
     * implementation may fill it and return that object.
     *
     * @param encodedData  The avro-encoded bytes of the data payload.
     * @param writerSchema The schema that was used to encode the data.
     * @param readerSchema The schema that is expected by the reader.
     * @param reuse        An optional object to be filled and returned to save on object construction
     *                     (may be null).
     * @return The decoded avro object.
     * @throws IOException If there is an error.
     */
    protected T decodeAvro(
        ByteBuffer encodedData,
        Schema writerSchema,
        Schema readerSchema,
        T reuse)
        throws IOException {
        final DatumReader<T> reader = createDatumReader(writerSchema, readerSchema);
        return reader.read(reuse,
            DecoderFactory.get().binaryDecoder(
                encodedData.array(),
                encodedData.position(),
                encodedData.limit() - encodedData.position(),
                null));
    }
}
